package org.qiyu.live.user.provider.service.impl;

import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.idea.qiyu.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import org.qiyu.live.user.constants.UserTagsEnum;
import org.qiyu.live.user.dto.UserCacheAsyncDeleteDTO;
import org.qiyu.live.user.dto.UserTagDTO;
import org.qiyu.live.user.provider.dao.mapper.IUserTagMapper;
import org.qiyu.live.user.provider.dao.po.UserTagPO;
import org.qiyu.live.user.provider.service.IUserTagService;
import org.qiyu.live.user.utils.TagInfoUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;
import org.qiyu.live.common.interfaces.utils.ConvertBeanUtils;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static org.qiyu.live.user.constants.CacheAsyncDeleteCode.USER_TAG_DELETE;
import static org.qiyu.live.user.constants.UserProviderTopicNames.CACHE_ASYNC_DELETE_TOPIC;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: sfx
 * Date: 2023-09-22
 * Time: 22:56
 */
@Service
public class UserTagServiceImpl implements IUserTagService {
    @Resource
    private IUserTagMapper userTagMapper;

    // 引入Redis缓存
    @Resource
    private RedisTemplate<String, UserTagDTO> redisTemplate;
    // 构建redis的key
    @Resource
    private UserProviderCacheKeyBuilder cacheKeyBuilder;

    // 引入MQ,用于延时双删
    @Resource
    private MQProducer mqProducer;

    private static final Logger log = LoggerFactory.getLogger(UserTagServiceImpl.class);

    /**
     * 如果用户已经设置过改标签,就不用设置了,直接返回失败
     *
     * @param userId      用户id
     * @param userTagEnum 用户标签
     * @return
     */
    @Override
    public boolean setTag(Long userId, UserTagsEnum userTagEnum) {
        if (userTagMapper.setTag(userId, userTagEnum.getFieldName(), userTagEnum.getTag()) > -1) {
            // 更新成功,删除缓存
            deleteUserTagKeyFromRedis(userId);
            return true;
        }
        // 我们还要考虑一种情况,我们必须事先有一个字段来进行运算,否则就会出现错误
        // 如果没有设置成功,说明就有两种情况,一种是已经有这个标签了,我真设置失败了
        UserTagPO userTagPO = userTagMapper.selectById(userId);
        if (userTagPO != null) {
            log.error("用户不存在 userId={}",userId);
            return false;
        }
        // 我们此时还要考虑高并发场景,多个线程穿插执行,设置标签,那么就很有可能插入的记录重复.
        // 因为我们设置的是主键,很有可能导致主键重复
        // 我们这里会使用set nx ex指令来保证线程安全,只会有一个线程插入成功,使用Redis,给MySQL分担压力,避免请求全部打入MySQL,造成访问压力

        // 我们使用execute把set nx ex 打包成一个原子操作
        String setNxKey = cacheKeyBuilder.buildUserTagLockKey(userId);
        redisTemplate.execute(new RedisCallback<String>() {
            @Override
            public String doInRedis(RedisConnection connection) throws DataAccessException {
                // 得到key和value的序列化器进行操作
                RedisSerializer keySerializer = redisTemplate.getKeySerializer();
                RedisSerializer valSerializer = redisTemplate.getValueSerializer();
                return (String) connection.execute("set", keySerializer.serialize(setNxKey),
                        valSerializer.serialize(-1),
                        "nx".getBytes(StandardCharsets.UTF_8),
                        "ex".getBytes(StandardCharsets.UTF_8),
                        "3".getBytes(StandardCharsets.UTF_8));
            }
        });
        // 一种是我没有这个标签字段,我第一次来,我需要插入一个记录,在设置标签
        userTagPO = new UserTagPO();
        userTagPO.setUserId(userId);
        userTagMapper.insert(userTagPO);
        // 在补偿一下设置标签
        return userTagMapper.setTag(userId, userTagEnum.getFieldName(), userTagEnum.getTag()) > 0;
    }

    @Override
    public boolean cancelTag(Long userId, UserTagsEnum userTagEnum) {
        // 取消标签后,我要删除缓存
        boolean updateStatus = userTagMapper.cancelTag(userId, userTagEnum.getFieldName(), userTagEnum.getTag()) > 0;
        deleteUserTagKeyFromRedis(userId);
        return updateStatus;
    }

    private void deleteUserTagKeyFromRedis(Long userId) {
        String key = cacheKeyBuilder.buildUserTagKey(userId);
        redisTemplate.delete(key);
        // 这里我们删除缓存,做一下优化,给MQ异步发送消息,延迟双删
        UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = new UserCacheAsyncDeleteDTO();
        userCacheAsyncDeleteDTO.setCode(USER_TAG_DELETE.getCode());
        Map<String,Object> jsonParams = new HashMap<>();
        jsonParams.put("userId",userId);
        userCacheAsyncDeleteDTO.setJson(JSON.toJSONString(jsonParams));
        Message message = new Message();
        // 设置topic
        message.setTopic(CACHE_ASYNC_DELETE_TOPIC);
        message.setBody(JSON.toJSONString(userCacheAsyncDeleteDTO).getBytes());
        message.setDelayTimeLevel(1);
        try {
            mqProducer.send(message);
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    @Override
    public boolean containsTag(Long userId, UserTagsEnum userTagEnum) {
        UserTagDTO userTagDTO = this.queryUserDTOFromRedis(userId);
        if (userTagDTO == null) {
            // 用户id不存在, userTagPO为null,直接返回
            return false;
        }
        if ("tag_info_01".equals(userTagEnum.getFieldName())) {
            return TagInfoUtils.tagMatchIng(userTagDTO.getTagInfo01(), userTagEnum.getTag());
        } else if ("tag_info_02".equals(userTagEnum.getFieldName())) {
            return TagInfoUtils.tagMatchIng(userTagDTO.getTagInfo02(), userTagEnum.getTag());
        } else if ("tag_info_03".equals(userTagEnum.getFieldName())) {
            return TagInfoUtils.tagMatchIng(userTagDTO.getTagInfo03(), userTagEnum.getTag());
        }
        return false;
    }

    private UserTagDTO queryUserDTOFromRedis(Long userId) {
        // 1. 查看redis中是否存在
        String key = cacheKeyBuilder.buildUserTagKey(userId);
        UserTagDTO userTagDTO = redisTemplate.opsForValue().get(key);
        // 2. redis中有,直接返回
        if (userTagDTO != null) {
            return userTagDTO;
        }
        // 3. 没有查询数据库,数据库没有直接返回
        UserTagPO userTagPO = userTagMapper.selectById(userId);
        if (userTagPO == null) {
            return null;
        }
        // 4. 数据库有,放到redis中
        userTagDTO = ConvertBeanUtils.convert(userTagPO, UserTagDTO.class);
        redisTemplate.opsForValue().set(key, userTagDTO);
        redisTemplate.expire(key,30, TimeUnit.MINUTES);
        // 5. 返回
        return userTagDTO;
    }
}